package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

import static com.heima.common.constants.HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC;

/**
 * @author JianKang
 * @create 2023-11-04 20:59
 */
@Configuration
@Slf4j
public class HotArticleStreamHandler {
    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //接受消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        //聚合流式处理
        stream.map((k, v) -> {
                    log.info("k={},v={}", k, v);
                    UpdateArticleMess mess = JSON.parseObject(v, UpdateArticleMess.class);
                    return new KeyValue<>(mess.getArticleId() + "", mess.getType().name() + ":" + mess.getAdd());
                }).groupBy((k, v) -> k).windowedBy(TimeWindows.of(Duration.ofSeconds(10))).aggregate(() -> "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0", (key, value, aggValue) -> {
                    if (StringUtils.isBlank(value)) {
                        return aggValue;
                    }
                    String[] aggAry = aggValue.split(",");
                    int col = 0, com = 0, like = 0, view = 0;
                    for (String agg : aggAry) {
                        String[] split = agg.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                            case COLLECTION:
                                col = Integer.parseInt(split[1]);
                                break;
                            case LIKES:
                                like = Integer.parseInt(split[1]);
                                break;
                            case COMMENT:
                                com = Integer.parseInt(split[1]);
                                break;
                            case VIEWS:
                                view = Integer.parseInt(split[1]);
                                break;
                        }
                    }

                    String[] valArr = value.split(":");
                    switch (UpdateArticleMess.UpdateArticleType.valueOf(valArr[0])) {
                        case COLLECTION:
                            col += Integer.parseInt(valArr[1]);
                            break;
                        case LIKES:
                            like += Integer.parseInt(valArr[1]);
                            break;
                        case COMMENT:
                            com += Integer.parseInt(valArr[1]);
                            break;
                        case VIEWS:
                            view += Integer.parseInt(valArr[1]);
                            break;
                    }
                    String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, like, view);
                    System.out.println("文章的ID" + key);
                    System.out.println("当前时间窗口内的消息处理结果：" + formatStr);
                    return formatStr;
                }, Materialized.as("hot-article-stream-count-001")).toStream()
                .map((k, v) -> new KeyValue<>(k.key(), formatObj(k.key(), v)))
                .to(HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return stream;
    }

    public String formatObj(String articleId, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));
        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        String json = JSON.toJSONString(mess);
        log.info("聚合消息处理后的结果：{}", json);
        return json;
    }
}
